package com.ebupt.migu.music.push.controller;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

/**
 * @author liuyangyang@e-byte.com
 * @description: TODO
 * @date 2024/8/15 15:38
 */
@Slf4j
@RestController
@RequestMapping("/kafka")
public class KAFKAController {
    @Resource
    private RedisTemplate redisTemplate;

    @GetMapping("/redis")
    public void redis() {

        String str = "BQ20241225103749";

        // 构建键名
        String uidKey = str + "-uid";
        String msisdnKey = str + "-phone";

        // 先删除可能存在的旧键
        redisTemplate.delete(uidKey);
        redisTemplate.delete(msisdnKey);

        // 获取集合操作对象
        BoundSetOperations<String, String> uid = redisTemplate.boundSetOps(uidKey);
        BoundSetOperations<String, String> msisdn = redisTemplate.boundSetOps(msisdnKey);

        int count = 1000;
        List<String> phoneNumbers = generatePhoneNumbers(count);

        // 使用线程池进行并发写入
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        // 将电话号码列表分成多个子列表，每个子列表由一个线程处理
        int numThreads = Runtime.getRuntime().availableProcessors();
        int batchSize = phoneNumbers.size() / numThreads;

        for (int i = 0; i < numThreads; i++) {
            int startIndex = i * batchSize;
            int endIndex = (i == numThreads - 1) ? phoneNumbers.size() : (i + 1) * batchSize;

            List<String> subList = phoneNumbers.subList(startIndex, endIndex);

            executorService.execute(() -> {
                for (String p : subList) {
                    System.out.println(p);
                    uid.add(p);
                    msisdn.add(p);
                }
            });
        }

        // 关闭线程池，等待所有任务完成
        executorService.shutdown();
        while (!executorService.isTerminated()) {
            // 等待线程池中的任务完成
        }
    }

    private List<String> generatePhoneNumbers(int count) {
        List<String> phoneNumbers = new ArrayList<>();
        Random random = new Random();
        for (int i = 0; i < count; i++) {
            // 假设手机号以 130 开头，后面随机生成 8 位数字
            String phoneNumber = "130" + String.format("%08d", random.nextInt(100000000));
            phoneNumbers.add(phoneNumber);
        }
        return phoneNumbers;
    }

    @GetMapping("/redis/clean")
    public void clean() {
        Set<String> key = redisTemplate.keys("DX20241102193131203*");
        if (key != null && !key.isEmpty()) {
            log.info("类似的KEY总计有{}个", key.size());
            List<String> keyList = new ArrayList<>(key);
            keyList.stream().forEach(k -> {
                log.info(k);
                redisTemplate.delete(k);
            });
            log.info("OK");
        }
    }

    @GetMapping("/redis/insert")
    public void insert() {
        Long start = System.currentTimeMillis();
        String key = "TEST-INSERT-20250114";
        redisTemplate.delete(key);
        Long expireTime = 1000L * 60 * 60 * 24 * 90;
        BoundSetOperations setOperations = redisTemplate.boundSetOps(key);
        // 使用线程池进行并发写入
        for (int i = 0; i <= 999999; i++) {
            Random random = new Random();
            String phoneNumber = "130" + String.format("%08d", random.nextInt(100000000));
            setOperations.add(phoneNumber);
        }
        System.out.println(Math.toIntExact(setOperations.size()));
        redisTemplate.expire(key, expireTime, TimeUnit.MILLISECONDS);
        Long end = System.currentTimeMillis();
        System.out.println("start time: " + start);
        System.out.println("end time: " + end);
        System.out.println("spend time: " + (end - start));
    }
}
